/*
 * CloudClient.cpp
 *
 *  Created on: Jan 23, 2012
 *      Author: hduser
 */

#include<iostream>
#include<fstream>
#include<string>
#include "hdfs.h"
#include "Connection.h"
#include "CloudClient.h"
#include "SocketClient.h"
#include "CommonUtilities.h"

using namespace std;
using namespace utilities;
using namespace logging;
using namespace networklib;

namespace cloud {

CloudClient::CloudClient(Connection* conn) {
	cloudConn = conn;
	init();
}

CloudClient::CloudClient(char* hostname, int port_number) {
	cerr<<"Entered Constructor CloudClient"<<endl;
	cloudConn = new Connection(hostname, port_number);
	cloudConn->Open();
	init();
	cerr<<"Exiting Constructor CloudClient"<<endl;
}

CloudClient::~CloudClient() {
	if (cloudConn != NULL) {
		cloudConn->Close();
	}
}

/*********************************************************/
/* PUBLIC FUNCTIONS           							 */
/*********************************************************/

void CloudClient::init() {
	logger = new Logger();
}

hdfsFileInfo* CloudClient::getFileInfoFromHDFS(char* hdfsfileName) {
	try {

		if (cloudConn == NULL)
			return NULL;

		if (cloudConn->cloudFS == NULL)
			return NULL;

		hdfsFileInfo *fileInfo = NULL;
		fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName);
		return fileInfo;

	} catch(exception ex) {
		ExceptionHandler::processException(&ex, "CloudClient::getHDFSFileInfo", "Caught exception");
		return NULL;
	}
}

int CloudClient::writeFileToHDFS(char* localfileName, char* hdfsfileName) {

	try {

		if (cloudConn == NULL)
			return -1;

		if (cloudConn->cloudFS == NULL)
			return -2;

		size_t num_read_bytes;
		char* localbuffer;
		streampos currentpos = 0;

		ifstream fl(localfileName, ios::in | ios::binary);
		fl.seekg(0, ios::end );
		size_t filelength = fl.tellg();

		hdfsFile writeFile = hdfsOpenFile(cloudConn->cloudFS, hdfsfileName, O_WRONLY|O_CREAT, 0, 0, 0);

		if(!writeFile) {
			cerr<<"Failed to open "<<hdfsfileName<<" for writing!"<<endl;
			return -1;
		}

		FileUtilities* fu = new FileUtilities();

		while (currentpos < filelength) {

			//cout<<"currentpos: "<<currentpos<<endl;
			//cout<<"filelength: "<<filelength<<endl;

			if (filelength - currentpos < BLOCK_SIZE)
				num_read_bytes = filelength - currentpos;
			else
				num_read_bytes = BLOCK_SIZE;

			num_read_bytes = fu->readBytes(&fl, &localbuffer, currentpos, num_read_bytes);

			if (num_read_bytes == -1)
			{
				hdfsFlush(cloudConn->cloudFS, writeFile);
				hdfsCloseFile(cloudConn->cloudFS, writeFile);
				hdfsDelete(cloudConn->cloudFS, hdfsfileName);
				fl.close();
				return -1;
			}

			size_t num_written_bytes = hdfsWrite(cloudConn->cloudFS, writeFile, (void*)localbuffer, num_read_bytes);

			currentpos = currentpos + num_read_bytes;
		}

		if (hdfsFlush(cloudConn->cloudFS, writeFile)) {
			cerr<<"Failed to 'flush' "<<hdfsfileName<<endl;
			return -1;
		}
		hdfsCloseFile(cloudConn->cloudFS, writeFile);
		fl.close();
		return 0;
	} catch(exception ex) {
		return -1;
	}
}

int CloudClient::getFileFromHDFS(char*  hdfsfileName, char* localfilename) {

	try {

		if (cloudConn == NULL)
					return -1;

		if (cloudConn->cloudFS == NULL)
			return -2;

		int exists = hdfsExists(cloudConn->cloudFS, hdfsfileName);
		char* buffer;
		size_t num_read_bytes;

		if (exists)	{
			cerr<<"Failed to validate existence of: "<<hdfsfileName<<endl;
			exit(-1);
		}

		hdfsFileInfo *fileInfo = NULL;
		size_t hdfsfilesize;

		if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName)) != NULL) {
			hdfsfilesize = fileInfo->mSize;
			hdfsFreeFileInfo(fileInfo, 1);
		}
		else {
			cerr<<"Failed to determine size of: "<<hdfsfileName<<endl;
			exit(-1);
		}

		hdfsFile readFile = hdfsOpenFile(cloudConn->cloudFS, hdfsfileName, O_RDONLY, 0, 0, 0);
		if (!readFile) {
			cerr<<"Failed to open "<<hdfsfileName<<" for reading!"<<endl;
			exit(-1);
		}

		size_t seekPos = 0;
		size_t currentPos = 0;

		seekPos = currentPos;
		if(hdfsSeek(cloudConn->cloudFS, readFile, seekPos)) {
			cerr<< "Failed to seek "<<hdfsfileName<<" for reading!"<<endl;
			exit(-1);
		}

		if((currentPos = hdfsTell(cloudConn->cloudFS, readFile)) != seekPos) {
			cerr<<"Failed to get current file position correctly! Got "<<currentPos<<endl;
			exit(-1);
		}

		ofstream fl(localfilename, ios::out | ios::binary);
		FileUtilities* fu = new FileUtilities();

		while (currentPos < hdfsfilesize) {

			//cout<<"currentPos: "<<currentPos<<endl;
			//cout<<"hdfsfilesize: "<<hdfsfilesize<<endl;

			if (hdfsfilesize - currentPos < BLOCK_SIZE)
				num_read_bytes = hdfsfilesize - currentPos;
			else
				num_read_bytes = BLOCK_SIZE;

			buffer = new char[num_read_bytes];
			num_read_bytes = hdfsPread(cloudConn->cloudFS, readFile, currentPos, (void*)buffer,num_read_bytes);

			if (num_read_bytes == -1)
			{
				hdfsCloseFile(cloudConn->cloudFS, readFile);
				fl.close();
				return -1;
			}

			fu->writeBytes(&fl, buffer, num_read_bytes);

			currentPos = currentPos + num_read_bytes;

			/*
			seekPos = currentPos;
			if(hdfsSeek(cloudConn->cloudFS, readFile, seekPos)) {
				cerr<< "Failed to seek "<<hdfsfileName<<" for reading!"<<endl;
				exit(-1);
			}

			if((currentPos = hdfsTell(cloudConn->cloudFS, readFile)) != seekPos) {
				cerr<<"Failed to get current file position correctly! Got "<<currentPos<<endl;
				exit(-1);
			}
			*/
		}

		hdfsCloseFile(cloudConn->cloudFS, readFile);
		fl.close();
		return 0;

	} catch(exception ex) {
		return -1;
	}
}

int CloudClient::getChunkFromHDFS(char*  hdfsfileName, int chunkIndex, int totalChunks, char** returnBuffer) {

	try {

		if (cloudConn == NULL)
			return -2; // connection error

		if (cloudConn->cloudFS == NULL)
			return -2; // connection error

		if (chunkIndex > totalChunks ) {
			cerr<<"chunk Index should be less than total chunks: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "chunk Index should be less than total chunks", hdfsfileName);
			return -11; // custom error
		}

		int exists = hdfsExists(cloudConn->cloudFS, hdfsfileName);

		if (exists)	{
			cerr<<"file not found in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "file not found in cloud", hdfsfileName);
			return -3; // file not exists
		}

		hdfsFileInfo *fileInfo = NULL;
		size_t hdfsfilesize;

		if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName)) != NULL) {
			hdfsfilesize = fileInfo->mSize;
			hdfsFreeFileInfo(fileInfo, 1);
		}
		else {
			cerr<<"cannot deterine size of the file in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "cannot deterine size of the file in cloud", hdfsfileName);
			return -3; // cannot determine file size
		}

		char* buffer;
		size_t num_read_bytes;

		hdfsFile readFile = hdfsOpenFile(cloudConn->cloudFS, hdfsfileName, O_RDONLY, 0, 0, 0);
		if (!readFile) {
			cerr<<"Failed to open "<<hdfsfileName<<" for reading!"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "cannot open file in cloud", hdfsfileName);
			return -3; // cannot open file
		}

		size_t seekPos = 0;
		size_t currentPos = chunkIndex * CHUNK_SIZE;

		seekPos = currentPos;
		if(hdfsSeek(cloudConn->cloudFS, readFile, seekPos)) {
			cerr<< "Failed to seek "<<hdfsfileName<<" for reading!"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "Failed to seek file in cloud", hdfsfileName);
			return -3; // seek error
		}

		if((currentPos = hdfsTell(cloudConn->cloudFS, readFile)) != seekPos) {
			cerr<<"Failed to get current file position correctly! Got "<<currentPos<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "failed to get current file position correctly", hdfsfileName);
			return -3; // seek error
		}

		//ofstream fl(localfilename, ios::out | ios::binary);
		//FileUtilities* fu = new FileUtilities();

		size_t maxCount = (((chunkIndex + 1) * CHUNK_SIZE) < hdfsfilesize) ? ((chunkIndex + 1) * CHUNK_SIZE) : hdfsfilesize;

		//char* returnBuffer; // = (char*)malloc(BLOCK_SIZE);
		size_t returnBufferSize = 0;

		while (currentPos < maxCount) {

			//cout<<"currentPos: "<<currentPos<<endl;
			//cout<<"hdfsfilesize: "<<hdfsfilesize<<endl;

			if (maxCount - currentPos < BLOCK_SIZE)
				num_read_bytes = maxCount - currentPos;
			else
				num_read_bytes = BLOCK_SIZE;

			buffer = new char[num_read_bytes];
			num_read_bytes = hdfsPread(cloudConn->cloudFS, readFile, currentPos, (void*)buffer,num_read_bytes);

			if (num_read_bytes == -1)
			{
				hdfsCloseFile(cloudConn->cloudFS, readFile);
				//fl.close();
				logger->logMessage(NULL, NULL, NULL, "CloudClient::getChunkFromHDFS", "read failed", hdfsfileName);
				return -3; // file error
			}

			size_t tempPos = returnBufferSize;
			returnBufferSize += num_read_bytes;
			*returnBuffer = (char*)realloc((void*)*returnBuffer, returnBufferSize);
			memcpy(*returnBuffer + tempPos, buffer, num_read_bytes);

			//fu->writeBytes(&fl, buffer, num_read_bytes);

			currentPos = currentPos + num_read_bytes;
		}

		hdfsCloseFile(cloudConn->cloudFS, readFile);
		//fl.close();
		return 0;

	} catch(exception ex) {
		ExceptionHandler::processException(&ex, "CloudClient::getChunkFromHDFS", "Caught exception");
		return -1;
	}
}

//size_t CloudClient::processChunk(chunkargs* args) {
size_t CloudClient::processChunk(int VendorID, char* hdfsfileName, int chunkIndex, int totalChunks, char* edgeserver_host, int edgeserver_port) {
	try {

		cout<<"VendorID = "<<VendorID<<endl;
		cout<<"chunkIndex = "<<chunkIndex<<endl;
		cout<<"edgeserver_host = "<<edgeserver_host<<endl;
		cout<<"edgeserver_port = "<<edgeserver_port<<endl;
		cout<<"hdfsfileName = "<<hdfsfileName<<endl;
		cout<<"totalChunks = "<<totalChunks<<endl;


		if (cloudConn == NULL)
			return -2; // connection error

		if (cloudConn->cloudFS == NULL)
			return -2; // connection error

		if (chunkIndex > totalChunks ) {
			cerr<<"chunk Index should be less than total chunks: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "chunk Index should be less than total chunks", hdfsfileName);
			return -11; // custom error
		}

		cout<<"pc 1"<<endl;

		int exists = hdfsExists(cloudConn->cloudFS, hdfsfileName);

		if (exists)	{
			cerr<<"file not found in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "file not found in cloud", hdfsfileName);
			return -3; // file not exists
		}

		cout<<"pc 2"<<endl;

		hdfsFileInfo *fileInfo = NULL;
		size_t hdfsfilesize;

		if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName)) != NULL) {
			hdfsfilesize = fileInfo->mSize;
			hdfsFreeFileInfo(fileInfo, 1);
		}
		else {
			cerr<<"cannot deterine size of the file in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "cannot deterine size of the file in cloud", hdfsfileName);
			return -3; // cannot determine file size
		}

		cout<<"pc 3"<<endl;

		char* buffer;
		size_t num_read_bytes;

		hdfsFile readFile = hdfsOpenFile(cloudConn->cloudFS, hdfsfileName, O_RDONLY, 0, 0, 0);
		if (!readFile) {
			cerr<<"Failed to open "<<hdfsfileName<<" for reading!"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "cannot open file in cloud", hdfsfileName);
			return -3; // cannot open file
		}

		cout<<"pc 4"<<endl;

		size_t seekPos = 0;
		size_t currentPos = (chunkIndex -1) * CHUNK_SIZE;

		seekPos = currentPos;
		if(hdfsSeek(cloudConn->cloudFS, readFile, seekPos)) {
			cerr<< "Failed to seek "<<hdfsfileName<<" for reading!"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to seek file in cloud", hdfsfileName);
			return -3; // seek error
		}

		if((currentPos = hdfsTell(cloudConn->cloudFS, readFile)) != seekPos) {
			cerr<<"Failed to get current file position correctly! Got "<<currentPos<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "failed to get current file position correctly", hdfsfileName);
			return -3; // seek error
		}

		cout<<"pc 5"<<endl;

		//ofstream fl(localfilename, ios::out | ios::binary);
		//FileUtilities* fu = new FileUtilities();

		// connect to server
		char* service = new char[5];
		sprintf(service, "%d", edgeserver_port);
		cout<<"service = "<<service;

		//return -1;

		SocketClient* client = new SocketClient(edgeserver_host, service);
		if (!client->connected) {
			cerr<<"Failed to connect to edge server"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to connect to edge server", hdfsfileName);
			return -4; // socket communication error
		}

		cout<<"pc 6"<<endl;

		int ibyteCount;
		char* socketRecvBuffer;

		// send file send command CMD_ACCEPT
		ibyteCount = client->sendData((char*)CMD_ACCEPT, CMD_LENGTH);

		if (ibyteCount != CMD_LENGTH) {
			cerr<<"Failed to send ACCEPT command to edge server"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to send ACCEPT command to edge server", hdfsfileName);
			client->disconnect();
			return -4; // socket communication error
		}

		//TODO if required:  wait for ack ACK_ACCEPTED
		socketRecvBuffer = (char*)malloc(CMD_LENGTH);
		ibyteCount = client->receiveData(socketRecvBuffer, CMD_LENGTH);

		if (ibyteCount != CMD_LENGTH) {
			cerr<<"Failed to receive ACCEPT ack from edge server"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to receive ACCEPT ack from edge server", hdfsfileName);
			client->disconnect();
			return -4; // socket communication error
		}

		cout<<"ACK_ACCEPTED - "<<socketRecvBuffer<<endl;

		// form file name
		//char* fullFilePath; // = (char*)malloc(FILENAME_SIZE);

		/*
		string s;
		stringstream out;
		out << VendorID;
		s = out.str();

		fullFilePath = strcpy(fullFilePath, "/");
		fullFilePath = strcpy(fullFilePath, (char*)s);
		fullFilePath = strcpy(fullFilePath, "/");
		fullFilePath = strcat(fullFilePath, hdfsfileName);
		fullFilePath = strcat(fullFilePath, "_");
		fullFilePath = strcat(fullFilePath, chunkIndex);
		*/

		CommonUtilities* cu = new CommonUtilities();

		string sfullFilePath;
		sfullFilePath.append(cu->to_string(VendorID));
		sfullFilePath.append(",");
		sfullFilePath.append(hdfsfileName);
		sfullFilePath.append(",");
		sfullFilePath.append(cu->to_string(chunkIndex));

		//sprintf(fullFilePath, "%d,%s,%d",VendorID, hdfsfileName, chunkIndex);

		char* fullFilePath = (char*)sfullFilePath.c_str();

		// send file name
		ibyteCount = client->sendData(fullFilePath, strlen(fullFilePath));

		if (ibyteCount != strlen(fullFilePath)) {
			cerr<<"Failed to send FILE NAME and PATH to edge server"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to send FILE NAME and PATH to edge server", hdfsfileName);
			client->disconnect();
			return -4; // socket communication error
		}

		//TODO: if required, wait for ready ack ACK_READY
		socketRecvBuffer = (char*)malloc(CMD_LENGTH);
		ibyteCount = client->receiveData(socketRecvBuffer, CMD_LENGTH);

		if (ibyteCount != CMD_LENGTH) {
			cerr<<"Failed to receive READY command from edge server"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to receive READY command from edge server", hdfsfileName);
			client->disconnect();
			return -4; // socket communication error
		}

		cout<<"ACK_READY - "<<socketRecvBuffer<<endl;

		size_t maxCount = (((chunkIndex) * CHUNK_SIZE) < hdfsfilesize) ? ((chunkIndex) * CHUNK_SIZE) : hdfsfilesize;

		//char* returnBuffer; // = (char*)malloc(BLOCK_SIZE);
		size_t returnBufferSize = 0;

		while (currentPos < maxCount) {

			//cout<<"currentPos: "<<currentPos<<endl;
			//cout<<"hdfsfilesize: "<<hdfsfilesize<<endl;

			if (maxCount - currentPos < BLOCK_SIZE)
				num_read_bytes = maxCount - currentPos;
			else
				num_read_bytes = BLOCK_SIZE;

			buffer = new char[num_read_bytes];
			num_read_bytes = hdfsPread(cloudConn->cloudFS, readFile, currentPos, (void*)buffer,num_read_bytes);

			if (num_read_bytes == -1)
			{
				hdfsCloseFile(cloudConn->cloudFS, readFile);
				// send ERROR message
				//client->sendData((char*)CMD_ERROR, strlen(CMD_ERROR));
				// close server connection
				client->disconnect();
				//fl.close();
				logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "read failed", hdfsfileName);
				return -3; // file error
			}

			size_t tempPos = returnBufferSize;
			returnBufferSize += num_read_bytes;
			//*returnBuffer = (char*)realloc((void*)*returnBuffer, returnBufferSize);
			//memcpy(*returnBuffer + tempPos, buffer, num_read_bytes);

			//fu->writeBytes(&fl, buffer, num_read_bytes);

			// send to server
			int iTotalBytes = 0;
			int iCurrentPos = 0;

			while (iTotalBytes < num_read_bytes) {

				if (num_read_bytes - iCurrentPos < TCP_BUFFER_SIZE)
					ibyteCount = num_read_bytes - iCurrentPos;
				else
					ibyteCount = TCP_BUFFER_SIZE;

				iCurrentPos = iTotalBytes;
				ibyteCount = client->sendData(buffer+iCurrentPos, ibyteCount);

				cout<<"ibyteCount = "<<ibyteCount<<endl;
				//cout<<"num_read_bytes = "<<num_read_bytes<<endl;
				cout<<"errorno = "<<errno<<endl;
				cout<<"error des = "<<strerror(errno)<<endl;

				if (ibyteCount < 0) {
					cerr<<"Failed to send data to edge server"<<endl;
					logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to send data to edge server", hdfsfileName);
					client->disconnect();
					return -4; // socket communication error
				}

				iTotalBytes+=ibyteCount;
			}

			currentPos = currentPos + num_read_bytes;
		}

		hdfsCloseFile(cloudConn->cloudFS, readFile);

		socketRecvBuffer = (char*)malloc(CMD_LENGTH);
		ibyteCount = client->receiveData(socketRecvBuffer, CMD_LENGTH);

		if (ibyteCount != CMD_LENGTH) {
			cerr<<"Failed to receive READY command for SUCCESS from edge server"<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to receive READY command for SUCCESS from edge server", hdfsfileName);
			client->disconnect();
			return -4; // socket communication error
		}

		// send SUCCESS message
		client->sendData((char*)CMD_SUCCESS, CMD_LENGTH);

		// close server connection
		client->disconnect();

		//fl.close();
		return returnBufferSize;

	} catch(exception ex) {
		ExceptionHandler::processException(&ex, "CloudClient::processChunk", "Caught exception");
		return -1;
	}
}

int CloudClient::calculateNumberOfChunks(char*  hdfsfileName) {

	try {

		if (cloudConn == NULL)
			return -1;

		if (cloudConn->cloudFS == NULL)
			return -2;

		int exists = hdfsExists(cloudConn->cloudFS, hdfsfileName);

		if (exists)	{
			cerr<<"file not found in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::calculateNumberOfChunks", "file not found in cloud", hdfsfileName);
			return -3; // file not exists
		}

		hdfsFileInfo *fileInfo = NULL;
		size_t hdfsfilesize;

		if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, hdfsfileName)) != NULL) {
			hdfsfilesize = fileInfo->mSize;
			hdfsFreeFileInfo(fileInfo, 1);
		}
		else {
			cerr<<"cannot deterine size of the file in cloud: "<<hdfsfileName<<endl;
			logger->logMessage(NULL, NULL, NULL, "CloudClient::calculateNumberOfChunks", "cannot deterine size of the file in cloud", hdfsfileName);
			return -4; // cannot determine file size
		}

		cout<<"hdfsfilesize = "<<hdfsfilesize<<endl;
		return (int)(hdfsfilesize / CHUNK_SIZE) + 1;

	} catch(exception ex) {
		ExceptionHandler::processException(&ex, "CloudClient::calculateNumberOfChunks", "Caught exception");
		return -1;
	}
}

int CloudClient::scanCloud(char* sVendorID, char* cloudroot, char* mainserver_host, char* mainserver_port, char* cloudserver_host, char* cloudserver_port) {
	try {

			if (cloudConn == NULL)
				return -1;

			if (cloudConn->cloudFS == NULL)
				return -2;

			int exists = hdfsExists(cloudConn->cloudFS, cloudroot);

			if (exists)	{
				cerr<<"file not found in cloud: "<<cloudroot<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::scanCloud", "path not found in cloud", cloudroot);
				return -3; // file not exists
			}

			hdfsFileInfo *fileInfo = NULL;

			if((fileInfo = hdfsGetPathInfo(cloudConn->cloudFS, cloudroot)) != NULL) {

				//cout<<"Name: "<<fileInfo->mName<<endl;
				if (fileInfo->mKind == 'D')
				{
					cout<<"Directory: "<<fileInfo->mName<<endl;
					hdfsFileInfo *fileList = 0;
					int numEntries = 0;
					if((fileList = hdfsListDirectory(cloudConn->cloudFS, cloudroot, &numEntries)) != NULL) {
						int i = 0;
						for(i=0; i < numEntries; ++i) {
							scanCloud(sVendorID, fileList[i].mName, mainserver_host, mainserver_port, cloudserver_host, cloudserver_port);
						}
						hdfsFreeFileInfo(fileList, numEntries);
					} else {
						if (errno) {
							return -1;
						} else {
							return 1;
						}
					}
				}
				else
				{
					cout<<"Process File: "<<fileInfo->mName<<endl;

					string sCloudAddress = "hdfs://";
					sCloudAddress.append(cloudserver_host);
					sCloudAddress.append(":");
					sCloudAddress.append(cloudserver_port);

					string sCloudFileName(fileInfo->mName);

					CommonUtilities* cu = new CommonUtilities();

					sCloudFileName = cu->replace(sCloudFileName, sCloudAddress, "");

					char* processFileName = (char*)sCloudFileName.c_str();

					SocketClient* client = new SocketClient(mainserver_host, mainserver_port);

					if (!client->connected) {
						cerr<<"Failed to connect to edge server"<<endl;
						logger->logMessage(NULL, NULL, NULL, "CloudClient::scanCloud", "Failed to connect to main server", processFileName);
						return -4; // socket communication error
					}


					int ibyteCount;
					char* socketRecvBuffer;

					// send file send command CMD_NEWDATA
					ibyteCount = client->sendData((char*)CMD_NEWDATA, CMD_LENGTH);

					if (ibyteCount != CMD_LENGTH) {
						cerr<<"Failed to send NEWDATA command to main server"<<endl;
						logger->logMessage(NULL, NULL, NULL, "CloudClient::scanCloud", "Failed to send NeWDATA command to main server", processFileName);
						client->disconnect();
						return -4; // socket communication error
					}

					//TODO if required:  wait for ack ACK_ACCEPTED
					socketRecvBuffer = (char*)malloc(CMD_LENGTH);
					ibyteCount = client->receiveData(socketRecvBuffer, CMD_LENGTH);

					if (ibyteCount != CMD_LENGTH) {
						cerr<<"Failed to receive ACCEPT ack from main server"<<endl;
						logger->logMessage(NULL, NULL, NULL, "CloudClient::scanCloud", "Failed to receive ACCEPT ack from main server", processFileName);
						client->disconnect();
						return -4; // socket communication error
					}

					cout<<"ACK_ACCEPTED - "<<socketRecvBuffer<<endl;

					string sfullFilePath;
					sfullFilePath.append(processFileName);
					sfullFilePath.append(",");
					sfullFilePath.append(sVendorID);

					char* fullFilePath = (char*)sfullFilePath.c_str();

					// send file name
					ibyteCount = client->sendData(fullFilePath, strlen(fullFilePath));

					if (ibyteCount != strlen(fullFilePath)) {
						cerr<<"Failed to send FILE NAME and PATH to main server"<<endl;
						logger->logMessage(NULL, NULL, NULL, "CloudClient::scanCloud", "Failed to send FILE NAME and PATH to main server", processFileName);
						client->disconnect();
						return -4; // socket communication error
					}
				}
				hdfsFreeFileInfo(fileInfo, 1);
			}
			else {
				cerr<<"cannot cannot scan the path in cloud: "<<cloudroot<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::scanCloud", "cannot cannot scan the path in cloud", cloudroot);
				return -4; // cannot scan the path in cloud
			}

			return 1;

		} catch(exception ex) {
			ExceptionHandler::processException(&ex, "CloudClient::scanCloud", "Caught exception");
			return -1;
		}
}

int CloudClient::deleteFileFolderfromEdgeServer(int VendorID,char * fileName,int chunkIndex, char* edgeserver_host, int edgeserver_port)
{
	try {

			cout<<"VendorID = "<<VendorID<<endl;
			cout<<"chunkIndex = "<<chunkIndex<<endl;
			cout<<"edgeserver_host = "<<edgeserver_host<<endl;
			cout<<"edgeserver_port = "<<edgeserver_port<<endl;
			cout<<"FileName = :"<<fileName<<endl;

			// connect to server
			char* service = new char[5];
			sprintf(service, "%d", edgeserver_port);
			cout<<"service = "<<service;


			SocketClient* client = new SocketClient(edgeserver_host, service);
			if (!client->connected)
			{
				cerr<<"Failed to connect to edge server"<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::deleteFileFolderfromEdgeServer", "Failed to connect to edge server", fileName);
				return -4; // socket communication error
			}

			cout<<"pc 6"<<endl;

			int ibyteCount;
			char* socketRecvBuffer;

			// send file send command CMD_ACCEPT
			ibyteCount = client->sendData((char*)CMD_DELETE, CMD_LENGTH);

			if (ibyteCount != CMD_LENGTH)
			{
				cerr<<"Failed to send DELETE command to edge server"<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::deleteFileFolderfromEdgeServer", "Failed to send ACCEPT command to edge server", fileName);
				client->disconnect();
				return -4; // socket communication error
			}

			//TODO if required:  wait for ack ACK_ACCEPTED
			socketRecvBuffer = (char*)malloc(CMD_LENGTH);
			ibyteCount = client->receiveData(socketRecvBuffer, CMD_LENGTH);

			if (ibyteCount != CMD_LENGTH) {
				cerr<<"Failed to receive ACCEPT ack from edge server"<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::deleteFileFolderfromEdgeServer", "Failed to receive ACCEPT ack from edge server", fileName);
				client->disconnect();
				return -4; // socket communication error
			}

			cout<<"ACK_FOR_DELETE_ACCEPTED - "<<socketRecvBuffer<<endl;

			// form file name
			//char* fullFilePath; // = (char*)malloc(FILENAME_SIZE);

			/*
			string s;
			stringstream out;
			out << VendorID;
			s = out.str();

			fullFilePath = strcpy(fullFilePath, "/");
			fullFilePath = strcpy(fullFilePath, (char*)s);
			fullFilePath = strcpy(fullFilePath, "/");
			fullFilePath = strcat(fullFilePath, hdfsfileName);
			fullFilePath = strcat(fullFilePath, "_");
			fullFilePath = strcat(fullFilePath, chunkIndex);
			*/

			CommonUtilities* cu = new CommonUtilities();

			string sfullFilePath;
			sfullFilePath.append(cu->to_string(VendorID));
			sfullFilePath.append(",");
			sfullFilePath.append(fileName);
			sfullFilePath.append(",");
			sfullFilePath.append(cu->to_string(chunkIndex));

			//sprintf(fullFilePath, "%d,%s,%d",VendorID, hdfsfileName, chunkIndex);

			char* fullFilePath = (char*)sfullFilePath.c_str();

			// send file name,vendorID and chunkindex comma separated string
			ibyteCount = client->sendData(fullFilePath, strlen(fullFilePath));

			if (ibyteCount != strlen(fullFilePath)) {
				cerr<<"Failed to send FILE NAME and PATH to edge server"<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::deleteFileFolderfromEdgeServer", "Failed to send FILE NAME and PATH, vendor ID and  to edge serve chunk index to Edge server", fileName);
				client->disconnect();
				return -4; // socket communication error
			}

			//TODO: if required, wait for ready ack ACK_READY
			socketRecvBuffer = (char*)malloc(CMD_LENGTH);
			ibyteCount = client->receiveData(socketRecvBuffer, CMD_LENGTH);

			if (ibyteCount != CMD_LENGTH) {
				cerr<<"Failed to receive SUCCESS DELETE  from edge server"<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to receive READY command from edge server",fileName);
				client->disconnect();
				return -4; // socket communication error
			}

			cout<<"ACK_READY - "<<socketRecvBuffer<<endl;
			/*
			size_t maxCount = (((chunkIndex) * CHUNK_SIZE) < hdfsfilesize) ? ((chunkIndex) * CHUNK_SIZE) : hdfsfilesize;

			//char* returnBuffer; // = (char*)malloc(BLOCK_SIZE);
			size_t returnBufferSize = 0;

			while (currentPos < maxCount) {

				//cout<<"currentPos: "<<currentPos<<endl;
				//cout<<"hdfsfilesize: "<<hdfsfilesize<<endl;

				if (maxCount - currentPos < BLOCK_SIZE)
					num_read_bytes = maxCount - currentPos;
				else
					num_read_bytes = BLOCK_SIZE;

				buffer = new char[num_read_bytes];
				num_read_bytes = hdfsPread(cloudConn->cloudFS, readFile, currentPos, (void*)buffer,num_read_bytes);

				if (num_read_bytes == -1)
				{
					hdfsCloseFile(cloudConn->cloudFS, readFile);
					// send ERROR message
					//client->sendData((char*)CMD_ERROR, strlen(CMD_ERROR));
					// close server connection
					client->disconnect();
					//fl.close();
					logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "read failed", hdfsfileName);
					return -3; // file error
				}

				size_t tempPos = returnBufferSize;
				returnBufferSize += num_read_bytes;
				//*returnBuffer = (char*)realloc((void*)*returnBuffer, returnBufferSize);
				//memcpy(*returnBuffer + tempPos, buffer, num_read_bytes);

				//fu->writeBytes(&fl, buffer, num_read_bytes);

				// send to server
				int iTotalBytes = 0;
				int iCurrentPos = 0;

				while (iTotalBytes < num_read_bytes) {

					if (num_read_bytes - iCurrentPos < TCP_BUFFER_SIZE)
						ibyteCount = num_read_bytes - iCurrentPos;
					else
						ibyteCount = TCP_BUFFER_SIZE;

					iCurrentPos = iTotalBytes;
					ibyteCount = client->sendData(buffer+iCurrentPos, ibyteCount);

					cout<<"ibyteCount = "<<ibyteCount<<endl;
					//cout<<"num_read_bytes = "<<num_read_bytes<<endl;
					cout<<"errorno = "<<errno<<endl;
					cout<<"error des = "<<strerror(errno)<<endl;

					if (ibyteCount < 0) {
						cerr<<"Failed to send data to edge server"<<endl;
						logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to send data to edge server", hdfsfileName);
						client->disconnect();
						return -4; // socket communication error
					}

					iTotalBytes+=ibyteCount;
				}

				currentPos = currentPos + num_read_bytes;
			}

			hdfsCloseFile(cloudConn->cloudFS, readFile);

			socketRecvBuffer = (char*)malloc(CMD_LENGTH);
			ibyteCount = client->receiveData(socketRecvBuffer, CMD_LENGTH);

			if (ibyteCount != CMD_LENGTH) {
				cerr<<"Failed to receive READY command for SUCCESS from edge server"<<endl;
				logger->logMessage(NULL, NULL, NULL, "CloudClient::processChunk", "Failed to receive READY command for SUCCESS from edge server", hdfsfileName);
				client->disconnect();
				return -4; // socket communication error
			}

			// send SUCCESS message
			client->sendData((char*)CMD_SUCCESS, CMD_LENGTH);

			// close server connection
			client->disconnect();

			//fl.close();
			return returnBufferSize;*/

			return 10;
		} catch(exception ex) {
			ExceptionHandler::processException(&ex, "CloudClient::deleteFileFolderfromEdgeServer", "Caught exception");
			return -1;
		}





}

void CloudClient::printMessage() {
	cout<<"hello from CloudClient"<<endl;
}

}
